package com.yuaer.demo.springsparkjobhandlerdemo.executor.processor;

import com.yuaer.demo.springsparkjobhandlerdemo.executor.GlobalContext;
import com.yuaer.demo.springsparkjobhandlerdemo.executor.NodeContext;
import org.apache.spark.sql.*;

import java.time.LocalDate;
import java.util.List;
import java.util.Map;

import static org.apache.spark.sql.functions.*;

public class ConsumerNodeProcessor implements NodeProcessor{

    @Override
    public Dataset<Row> process(GlobalContext context, NodeContext nodeContext) {
        Map<String, Object> params = nodeContext.getParams();
        String cardType = (String) params.get("cardType");

        Column amountCol = functions.col("sum_amount");
        Column countCol = functions.col("count_trans");

        long amount = (long) params.get("amount");

        SparkSession sparkSession = context.getSpark();
        //todo 需保证上游数据输出列一致
        Dataset<Row> batchResult = nodeContext.getInputs().values().stream()
                .reduce((d1, d2) -> d1.union(d2))
                .orElse(sparkSession.emptyDataFrame());
        /**
         * TODO: 2021/3/30 这里可能需要将LocalDate 转为字符串或java.sql.Date类型
         * 参考 java.sql.Date.valueOf(localDate)
         * 消费事件的主要逻辑均在此
         */
        return batchResult
                .filter(col("cardType").equalTo(cardType))
                .groupBy("custNo","date") //仅做示例
                .agg(
                        sum("amount").alias("regional_sales"),
                        countDistinct("date").alias("active_days")
                )
                .withColumn("result", amountCol.gt(amount).and(countCol.gt(10)))
                .withColumn("detail", functions.to_json(functions.struct(
                        amountCol.alias("amount"),
                        countCol.alias("count")
                )));
        /**
         *
         * 最终输出数据类似如下:
         * cust_no	act_no	sum_amount	count_trans	result	detail
         * 123	a	25000	123	true	{"amount":25000,"count":123}
         * 456	a	1200	32	false	{"amount":1200,"count":32}
         *
         *
         */
    }

}
